#!/usr/bin/env python

'''Unittest for the RtmAPI.

Far from complete, but it's a start.
'''

import unittest
import sys
import urllib
import StringIO
import exceptions
import logging

from pymock import PyMockTestCase

# Make sure the rtmapi module from the source distribution is used
sys.path.insert(0, '..')

import rtmapi
rtmapi.set_log_level(logging.FATAL)

print "Testing RtmAPI version %s" % rtmapi.__version__

# Some useful constants
EURO_UNICODE = u'\u20ac'
EURO_UTF8 = EURO_UNICODE.encode('utf-8')
U_UML_UNICODE = u'\u00fc'
U_UML_UTF8 = U_UML_UNICODE.encode('utf-8')

########################################################################
# Note for developers / testers:
# The key is a valid one, but the secret is NOT. It's just random junk 
# to test the signing methods. Please DO NOT abuse the key trying to do
# something real with this api_key/secret pair. It will NOT work, and
# you'll only get the api_key revoked by RTM.
#    Thanks !!!
key = 'c1fdacce78e33b4fcaf8a7904f6bd414'
secret = 'e8196e735aa469b0' 
########################################################################

logging.basicConfig()
LOG = logging.getLogger(__name__)

def python_version(major, minor, micro):
    '''Function decorator, skips calling the function when the python version
    is older than the given version.
    '''

    current_version = sys.version_info[0:3]

    def just_pass(*args, **kwargs):
        pass

    def decorator(method):
        if current_version < (major, minor, micro):
            LOG.warn('Skipping %s, Python version %s.%s.%s too old' %
                    ((method.func_name, ) + current_version))
            return just_pass
        return method

    return decorator

class SuperTest(PyMockTestCase):
    '''Superclass for unittests, provides useful methods.'''
    
    def setUp(self):
        super(SuperTest, self).setUp()
        self.r = rtmapi.RtmAPI(key, secret)
        self.r_noauth = rtmapi.RtmAPI(key)

        # Remove any unwanted tokens
        self.r.token_cache.forget()
        self.r_noauth.token_cache.forget()

    def assertUrl(self, expected_protocol, expected_host, expected_path,
                  expected_query_arguments, actual_url):
        '''Asserts that the 'actual_url' matches the given parts.'''
            
        # Test the URL part by part
        (urltype, rest) = urllib.splittype(actual_url)
        self.assertEqual(expected_protocol, urltype)
        
        (hostport, path) = urllib.splithost(rest)
        self.assertEqual(expected_host, hostport)
        
        (path, query) = urllib.splitquery(path)
        self.assertEqual(expected_path, path)
        
        attrvalues = query.split('&')
        attribs = dict(av.split('=') for av in attrvalues)
        self.assertEqual(expected_query_arguments, attribs)
    
class RtmApiTest(SuperTest):
    def test_repr(self):
        '''Class name and API key should be in repr output'''

        r = repr(self.r)
        self.assertTrue('RtmAPI' in r)
        self.assertTrue(key in r)

    def test_auth_url(self):
        '''Test the authentication URL generation'''
        
        args = dict(api_key=key, frob='frob', perms='read')
        args['api_sig'] = self.r.sign(args)
        
        url = self.r.auth_url(args['perms'], args['frob'])
        
        self.assertUrl('http', rtmapi.RtmAPI.rtm_host, 
                       rtmapi.RtmAPI.rtm_auth_form, args, 
                       url)
        
    def test_web_login_url(self):
        '''Test the web login URL.'''
        
        args = dict(api_key=key, perms='read')
        args['api_sig'] = self.r.sign(args)
        
        url = self.r.web_login_url(args['perms'])
        
        self.assertUrl('http', rtmapi.RtmAPI.rtm_host,
                       rtmapi.RtmAPI.rtm_auth_form, args,
                       url)
        
    def test_simple_echo(self):
        '''Test simple method call'''
        
        # We expect to get the parameters back
        xml = self.r.test.echo(foo=u'bar', spam=u'eggs')
        self.assertEqual(xml.find('foo').text, u'bar')
        self.assertEqual(xml.find('spam').text, u'eggs')
    
    def test_token_constructor(self):
        '''Test passing a token to the constructor'''
        
        token = '123-abc-def'
        
        # Pass the token
        rtm = rtmapi.RtmAPI(key, secret, token=token)
        
        # It should be in the in-memory token cache now
        self.assertEqual(token, rtm.token_cache.token)
        
        # But not in the on-disk token cache
        self.assertNotEqual(token, rtmapi.TokenCache(key))              

    def test_auth_token_without_secret(self):
        '''Auth tokens without secrets are meaningless'''

        
        token = '123-abc-def'
        
        # Create a normal RtmAPI object
        rtm = rtmapi.RtmAPI(key)

        rtm.token_cache.token = token
        self.assertRaises(exceptions.ValueError, rtm.test.echo,
                          foo='bar')

    def test_store_token(self):
        '''Tests that store_token=False RtmAPI uses SimpleTokenCache'''

        token_disk = '123-abc-disk'
        token_mem = '123-abc-mem'

        # Create a non-public-only instance, and set the on-disk token
        rtm = rtmapi.RtmAPI(key, secret)
        rtm.token_cache.token = token_disk
        
        rtm = rtmapi.RtmAPI(key, secret, store_token=False)

        # The token shouldn't be set
        self.assertEqual(None, rtm.token_cache.token)

        # Now set it
        rtm.token_cache.token = token_mem
        
        # It should not be in the on-disk token cache, only in memory
        self.assertEqual(token_disk, rtmapi.TokenCache(key).token)
        self.assertNotEqual(token_mem, rtmapi.TokenCache(key).token)

    def test_wrap_in_parser(self):
        '''Tests wrap_in_parser'''

        test = {'wrapped': False}

        def to_wrap(format, test_param):
            self.assertEqual('rest', format)
            self.assertEqual('test_value', test_param)
            test['wrapped'] = True

            return '<rst stat="ok"><element list_id="7" /></rst>'

        rst = self.r._RtmAPI__wrap_in_parser(to_wrap, parse_format='xmlnode',
                format='xmlnode', test_param='test_value')
        self.assertEqual('7', rst.element[0]['list_id'])
        self.assertTrue(test['wrapped'],
                        'Expected wrapped function to be called')

    def test_wrap_in_parser_no_format(self):
        '''Tests wrap_in_parser without a format in the wrapped arguments'''

        test = {'wrapped': False}

        def to_wrap(test_param):
            self.assertEqual('test_value', test_param)
            test['wrapped'] = True

            return '<rst stat="ok"><element list_id="7" /></rst>'

        rst = self.r._RtmAPI__wrap_in_parser(to_wrap, parse_format='xmlnode',
                test_param='test_value')
        self.assertEqual('7', rst.element[0]['list_id'])
        self.assertTrue(test['wrapped'],
                        'Expected wrapped function to be called')

class CachingTest(SuperTest):
    '''Tests that the caching framework works'''

    def test_cache_write(self):
        '''tests that the call result is written to cache'''

        dummy_id = '0123456789'
        cache_key = ('api_key=%s'
                     '&dummy_id=%s'
                     '&method=rtm.test.echo'
                     '&format=rest' % (key, dummy_id))
        
        r = rtmapi.RtmAPI(key, store_token=False, format='rest')
        r.cache = rtmapi.SimpleCache()
        self.assertEqual(0, len(r.cache))

        info = r.test.echo(dummy_id=dummy_id)

        self.assertEqual(info, r.cache.get(cache_key))

    def test_cache_read(self):
        '''Tests that cached data is returned if available'''

        dummy_id = '0123456789'
        cache_key = ('api_key=%s'
                     '&dummy_id=%s'
                     '&method=rtm.test.echo'
                     '&format=rest' % (key, dummy_id))
        faked_value = "FAKED_VALUE"
        
        r = rtmapi.RtmAPI(key, store_token=False, format='rest')
        r.cache = rtmapi.SimpleCache()
        r.cache.set(cache_key, faked_value)

        info = r.test.echo(dummy_id=dummy_id)

        self.assertEqual(faked_value, info)

    def test_cache_constructor_parameter(self):
        '''Tests that a cache is created when requested.'''

        r = rtmapi.RtmAPI(key, cache=True)
        self.assertNotEqual(None, r.cache, "Cache should not be None")

    # Test list of non-cacheable method calls

class FormatsTest(SuperTest):
    '''Tests the different parsed formats.'''

    @python_version(2, 5, 0)
    def test_etree_format_happy(self):
        '''Test ETree format'''

        etree = self.r_noauth.test.echo(dummy_id=u'eggs',
                    format='etree')
        self.assertEqual('xml.etree.ElementTree', etree.__module__)

    @python_version(2, 5, 0)
    def test_etree_format_error(self):
        '''Test ETree format in error conditions'''
 
        self.assertRaises(rtmapi.exceptions.RtmError,
                self.r_noauth.test.login, format='etree')

    def test_xmlnode_format(self):
        '''Test XMLNode format'''

        node = self.r_noauth.test.echo(dummy_id=u'eggs',
                    format='xmlnode')
        self.assertNotEqual(None, node.dummy_id[0])

    @python_version(2, 5, 0)
    def test_xmlnode_format_error(self):
        '''Test ETree format in error conditions'''
 
        self.assertRaises(rtmapi.exceptions.RtmError,
                self.r_noauth.test.login, format='xmlnode')
        
    @python_version(2, 5, 0)
    def test_xmlnode_default_format(self):
        '''Test setting the default format to xmlnode'''

        r = rtmapi.RtmAPI(key, format='xmlnode')
        xmlnode = r.test.echo(spam=u'eggs')
        self.assertEqual('rtmapi.xmlnode', xmlnode.__module__)

    def test_explicit_format(self):
        '''Test explicitly requesting a certain format'''
        
        xml = self.r_noauth.test.echo(foo=u'bar', format='rest')
        self.assertTrue(isinstance(xml, basestring))
        
        # Try to parse it
        rst = rtmapi.XMLNode.parse(xml, False)
        self.assertEqual(rst.foo[0].text, u'bar')
    
class SigningTest(SuperTest):
    '''Tests the signing of different arguments.'''

    def testSimple(self):
        '''Simple arguments, just ASCII'''
        
        signed = self.r.sign({'abc': 'def'})
        self.assertEqual('a6dc5cc7507be5793915dd3f59c4fb06', signed)

        # Order shouldn't matter
        signed = self.r.sign({'abc': 'def', 'foo': 'bar'})
        self.assertEqual('8eddbb2e6c58816a2ba2ab8b3936c20a', signed)

        signed = self.r.sign({'foo': 'bar', 'abc': 'def'})
        self.assertEqual('8eddbb2e6c58816a2ba2ab8b3936c20a', signed)

    def testUnicode(self):
        '''Test signing of Unicode data'''

        # Unicode can't be signed directly
        self.assertRaises(rtmapi.IllegalArgumentException, self.r.sign, {'abc': u'def'})

        # But converted to UTF-8 works just fine
        signed = self.r.sign({'abc': u'def'.encode('utf-8')})
        self.assertEqual('a6dc5cc7507be5793915dd3f59c4fb06', signed)
        
        # Non-ASCII data should work too
        data = EURO_UNICODE + U_UML_UNICODE
        signed = self.r.sign({'abc': data.encode('utf-8')})
        self.assertEqual('d3c05f8caf2808d3056061e8ec924f88', signed)

        # Straight UTF-8 should work too
        data = EURO_UTF8 + U_UML_UTF8
        signed = self.r.sign({'abc': data})
        self.assertEqual('d3c05f8caf2808d3056061e8ec924f88', signed)

class EncodingTest(SuperTest):
    '''Test URL encoding + signing of data. Tests using sets, because
    we don't know in advance in which order the arguments will show up,
    and we don't care about that anyway.
    '''
    
    def testSimple(self): 
        '''Test simple ASCII-only data'''

        encoded = self.r.encode_and_sign({'abc': 'def', 'foo': 'bar'})
        expected = set(['abc=def',
                        'foo=bar',
                        'api_sig=8eddbb2e6c58816a2ba2ab8b3936c20a'
                        ])
        self.assertEqual(expected, set(encoded.split('&')))

        # Order shouldn't matter for the signature
        encoded = self.r.encode_and_sign({'foo': 'bar', 'abc': 'def'})
        self.assertEqual(expected, set(encoded.split('&')))

    def testUnicode(self):
        '''Test Unicode data'''

        # Unicode strings with ASCII data only should result in the
        # same as in the testSimple() test. 
        encoded = self.r.encode_and_sign({'abc': u'def', 'foo': u'bar'})
        expected = set(['abc=def',
                        'foo=bar',
                        'api_sig=8eddbb2e6c58816a2ba2ab8b3936c20a'
                        ])
        self.assertEqual(expected, set(encoded.split('&')))

        # Non-ASCII UTF-8 data should work too
        # EURO = 0xE2 0x82 0xAC in UTF-8
        # U_UML = 0xC3 0xBC in UTF-8
        data = EURO_UNICODE + U_UML_UNICODE
        encoded = self.r.encode_and_sign({'abc': data.encode('utf-8')})
        expected = set(['abc=%E2%82%AC%C3%BC',
                        'api_sig=d3c05f8caf2808d3056061e8ec924f88'
                        ])
        self.assertEqual(expected, set(encoded.split('&')))

        # Straight Unicode should work too
        data = EURO_UNICODE + U_UML_UNICODE
        encoded = self.r.encode_and_sign({'abc': data})
        self.assertEqual(expected, set(encoded.split('&')))

    def testNoSecret(self):
        
        no_secret = rtmapi.RtmAPI(key)
        data = EURO_UNICODE + U_UML_UNICODE
        encoded = no_secret.encode_and_sign({'abc': data})
        self.assertEqual('abc=%E2%82%AC%C3%BC', encoded)

class DynamicMethodTest(SuperTest):
    '''Tests the dynamic methods used to interface with Remember The Milk.'''
    
    class FakeUrllib(object):
        '''Fake implementation of URLLib'''
    
        def __init__(self):
            self.data = None
            self.url = None
            
        def urlopen(self, url, postdata):
            self.url = url
            self.data = postdata
            
            return StringIO.StringIO('''<?xml version="1.0" encoding="utf-8"?>
                <rsp stat="ok"></rsp>''')

        def __getattr__(self, name):
            '''If we don't implement a method, call the original'''
            
            if not hasattr(urllib, name):
                raise AttributeError("No such attibute %s" % name)
            
            return getattr(urllib, name)
            
            #def original_caller(*args, **kwargs):
            #    original(*args, **kwargs)
            
    def setUp(self):
        super(DynamicMethodTest, self).setUp()

        # Set fake urllib
        self.fake_url_lib = self.FakeUrllib() 
        rtmapi.urllib = self.fake_url_lib

    def tearDown(self):
        super(DynamicMethodTest, self).tearDown()

        # Restore original urllib
        rtmapi.urllib = urllib
    
    def testUnicodeArgs(self):
        '''Tests whether Unicode arguments are properly handled.
        
        Tests using sets, since the order of the URL-encoded arguments
        can't be ensured.
        '''
        
        # Plain ASCII should work
        self.r.time.parse(text='tomorrow')
        sent = set(self.fake_url_lib.data.split('&'))
        expected = set(['api_key=%s' % key, 
                        'text=tomorrow', 
                        'method=rtm.time.parse', 
                        'api_sig=fe11d20b93c442fafb33276573078f31',
                        'format=rest'
                        ])
        self.assertEquals(expected, sent)
         
        # Unicode should work too
        self.r.time.parse(text=EURO_UNICODE+U_UML_UNICODE+' tomorrow')
        sent = set(self.fake_url_lib.data.split('&'))
        expected = set(['api_key=%s' % key,
                        'text=%E2%82%AC%C3%BC+tomorrow',
                        'method=rtm.time.parse',
                        'api_sig=8c3e91ba617637595147c060fd15526f',
                        'format=rest'
                        ])
        self.assertEquals(expected, sent)

    def test_private_attribute(self):
        '''Tests that we get an AttributeError when accessing an attribute
        starting with __.
        '''
        
        self.assertRaises(AttributeError, self.r, '__getList')

    def test_get_dynamic_method(self):
        '''Test the dynamic method calls on an RtmAPI instance.
        '''
        
        method = self.r.test.testMethod
        self.assertTrue(callable(method))
        self.assertEquals('rtm.test.testMethod', method._call_chain)

        # Override the standard RtmAPI call handler with a fake one
        # for testing purposes
        def test_handler(**args):
            return args
        method._call_handler = test_handler

        # Call the method and test the arguments
        rsp = method(spam='eggs', test=True)
        self.assertTrue(rsp.has_key('spam'))
        self.assertEquals(rsp['spam'], 'eggs')
        self.assertTrue(rsp.has_key('test'))
        self.assertTrue(rsp['test'])
        self.assertTrue(rsp.has_key('method'))
        self.assertEquals(rsp['method'], 'rtm.test.testMethod')

if __name__ == '__main__':
    unittest.main()
